package com.yun.storm.Sales;

import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Map;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

/**
 * 汇总SalesBolt多个线程的结果
 * 
 * @author WangXin
 *
 */
public class SalesDBWriteBolt extends BaseRichBolt {
	/**
	*
	*/
	private static final long serialVersionUID = 1L;
	private OutputCollector collector;
	private String tableName = "Sales";
//	private String databaseIP = "localhost";
//	private String databasePort = "3999";
//	private String databaseName = "jeeshop";
//	private String userName = "pentaho";
//	private String pwd = "pentaho";
	private Connection con = null;
	

	private String databaseIP = "10.14.1.60";
	private String databasePort = "3306";
	private String databaseName = "jeeshop";
	private String userName = "root";
	private String pwd = "Hitachi@123";

	@Override
	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
		this.collector = collector;

		try {
			// 第一步：加载MySQL的JDBC的驱动
			Class.forName("com.mysql.jdbc.Driver");
		} catch (ClassNotFoundException e) {
			System.out.println("Driver not found");
			e.printStackTrace();
		}

		try {
			// 第二步：创建与MySQL数据库的连接类的实例
			con = DriverManager.getConnection("jdbc:mysql://" + databaseIP + ":" + databasePort + "/" + databaseName,
					userName, pwd);
			con.prepareStatement("DROP TABLE IF EXISTS " + tableName).execute();

			StringBuilder createQuery = new StringBuilder("CREATE TABLE IF NOT EXISTS " + tableName + "(");
			createQuery.append("productID" + " varchar(255) NOT NULL,");
			createQuery.append("productName" + " varchar(255) default NULL,");
			createQuery.append("count" + " int(11) default NULL,");
			createQuery.append("countAmount" + " decimal(14,2) default NULL,");
//			createQuery.append("orderDate" + " datetime default NULL,");
			createQuery.append("PRIMARY KEY  (productID));");
			con.prepareStatement(createQuery.toString()).execute();
		} catch (SQLException e) {
			e.printStackTrace();
		}
	}

	@Override
	public void execute(Tuple input) {
		try {
			System.out.println("写入mysql方法");
			String productID = input.getStringByField("productID");
			System.out.println("#$#$#$#$#$#$# productID:"+productID);
			
			String productName = input.getStringByField("productName");
			System.out.println("#$#$#$#$#$#$# productName:"+productName);
			
			Integer count = input.getIntegerByField("count");
			System.out.println("#$#$#$#$#$#$# count:"+count);
			
			BigDecimal countAmount = BigDecimal.valueOf(input.getDoubleByField("countAmount"));
			System.out.println("#$#$#$#$#$#$# countAmount:"+countAmount);
			
			String orderDate = input.getStringByField("orderDate");
			System.out.println("#$#$#$#$#$#$# orderDate:"+orderDate);
			
			boolean insertFlg = input.getBooleanByField("insertFlg");
			System.out.println("#$#$#$#$#$#$# insertFlg:"+insertFlg);

			if (insertFlg) {
				StringBuilder insertQuery = new StringBuilder("INSERT INTO " + tableName + " VALUES (");
				insertQuery.append("'" + productID + "', ");
				insertQuery.append("'" + productName + "', ");
				insertQuery.append("'" + count + "', ");
				insertQuery.append("'" + countAmount + "')");
//				insertQuery.append("'" + orderDate + "')");
				Statement statement = con.createStatement();
				System.out.println("新增执行语句:"+insertQuery.toString());
				statement.executeUpdate(insertQuery.toString());
				System.out.println("向数据库插入"+productID+",销售量为:"+count + ",销售额为:"+countAmount);
			} else {
				StringBuilder updateQuery = new StringBuilder("UPDATE " + tableName + " SET ");
				updateQuery.append("count=" + "'" + count + "', ");
				updateQuery.append("countAmount=" + "'" + countAmount + "' ");
//				updateQuery.append("orderDate=" + "'" + orderDate + "' ");
				updateQuery.append("where productID =" + "'" + productID + "'");
				System.out.println(updateQuery);
				Statement statement = con.createStatement();
				System.out.println("更新执行语句:"+updateQuery.toString());
				statement.executeUpdate(updateQuery.toString());
				System.out.println("更新"+productID+",销售量为:"+count + ",销售额为:"+countAmount);
			}

			this.collector.ack(input);
		} catch (Exception e) {
			e.printStackTrace();
			this.collector.fail(input);
		}
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {

	}
}